package com.xinqing.bigdata.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * @Author:CHQ
  * @Date:2020 /6/28 14:58
  * @Description
  */
object WC {
  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("spark streaming wc")

    val streamingContext: StreamingContext = new StreamingContext(conf,Seconds(3))

    val inputStream: ReceiverInputDStream[String] = streamingContext.socketTextStream("localhost",7777)

    val resStream: DStream[(String, Int)] = inputStream.flatMap(_.split(",")).map((_,1)).reduceByKey(_+_)

    resStream.print()

    streamingContext.start()
    streamingContext.awaitTermination()
  }
}
